package com.study.business.log;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.study.business.log.bean.LogBean;
import com.study.business.log.mapper.LogMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Slf4j
@Service
public class LogService {

    private List<LogBean> logBeans = new ArrayList<>(50);

    @Value("${commit.batch.size:10}")
    private int commitBatchSize;

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @RabbitListener(queues = "log_queue", ackMode = "MANUAL", group = "Springboot")
    public void receiveLogAll(Message msg, Channel channel) throws Exception {
        byte[] body = msg.getBody();
        String msgJson = new String(body, StandardCharsets.UTF_8);
        LogBean logBean = JSON.parseObject(msgJson, LogBean.class);
        logBean.setCreateTime(new Date());
        if (null != logBean) {
            logBeans.add(logBean);
            if (logBeans.size() >= commitBatchSize) {
                flushToDb(logBeans);
                logBeans.clear();
                long deliveryTag = msg.getMessageProperties().getDeliveryTag();
                channel.basicAck(deliveryTag, true);
            }
        }
    }

    public void flushToDb(List<LogBean> logBeans) {

        SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
        LogMapper mapper = sqlSession.getMapper(LogMapper.class);
        logBeans.stream().forEach(logBean -> {
            try {
                mapper.add(logBean);
            } catch (Exception e) {
                log.info(JSON.toJSONString(logBean));
            }
        });
        sqlSession.commit();
        sqlSession.clearCache();
    }
}
